package com.bw.gmall.realtime.app.dwd;
/*
1.获取ods 中日志数topic_log
2.处理过滤topic_log
3.对新老用户进行处理
               is_news:1
                         状态=null  我要把当前人ts转换成年月日  跟新到状态值
                         状态！=null   判断当前数据的ts 和状态日期是否相同  如果不同跟新数据
               is_new:0
                         状态=null    将ts-1day 跟新到状态中
                         如果有值 不需要处理
4. 把主流数据拆分成5个流   分别存入到5个主题中
1.创建流式环境
2.设置并行度
3.运行jar 向topic_log主题 发送数据
4.从 Kafka 读取主流数据
5.数据清洗，转换结构
6.将脏数据写出到 Kafka 指定主题
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class BaseLogAppShopDwd {
    public static void main(String[] args) throws Exception{
            //todo 1.初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String topic = "ods_traffic";
        String groupid = "base_log_consumer";

        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupid));
        /*统计周期内买家访问您店铺（观看店铺自播直播间、
        观看自制全屏页短视频3秒及以上、
        浏览店铺自制图文3秒及以上、浏览全屏微详情、访问宝贝详情页及店铺其他页面*/
        source.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                if(jsonObject.getJSONObject("page")!=null){
                    String string = jsonObject.getJSONObject("page").getString("page_action");
                    if(string.equals("a")||string.equals("b")||string.equals("c")||string.equals("e")||string.equals("f")){
                        return true;
                    }
                    return false;
                }else{
                    return false;
                }

            }
        });

    }
}
